package com.wgzx.modbus.message;

import com.wgzx.modbus.constant.ModbusConstants;
import com.wgzx.modbus.entity.ModbusRsp;
import com.wgzx.modbus.entity.ModbusRspFrame;
import io.netty.util.concurrent.GlobalEventExecutor;
import lombok.extern.slf4j.Slf4j;
import reactor.core.publisher.Mono;
import reactor.core.publisher.MonoSink;
import reactor.core.scheduler.Schedulers;

import java.util.Map;
import java.util.concurrent.*;
import java.util.function.Consumer;

@Slf4j
public class DefaultMsgManager implements MsgManager {

    private final ScheduledExecutorService executorService = GlobalEventExecutor.INSTANCE;
    private final Map<Integer, MsgHolder> messageHolderStore;
    private final int globalResponseTimeout;

    public DefaultMsgManager() {
        this(ModbusConstants.SYNC_RESPONSE_TIMEOUT);
    }

    public DefaultMsgManager(int globalResponseTimeout) {
        this.globalResponseTimeout = globalResponseTimeout;
        this.messageHolderStore = new ConcurrentHashMap<>();
    }

    @Override
    public <T extends ModbusRsp> Mono<T> getResponseAsync(final int transactionId, Callback callback) {
        MsgHolder msgHolder = new MsgHolder(transactionId, getGlobalResponseTimeout());
        messageHolderStore.putIfAbsent(transactionId, msgHolder);
        msgHolder.setTimeoutFuture(executorService, () -> responseTimeout(transactionId));
        return Mono.create((Consumer<MonoSink<T>>) sink -> {
            try {
                msgHolder.setListener(new SinkMsgListener(sink));
                callback.run();
            } catch (Exception e) {
                sink.error(e);
            }
        }).publishOn(Schedulers.parallel());
    }

    @Override
    public <T extends ModbusRsp> CompletableFuture<T> getResponseSync(final int transactionId) {
        MsgHolder msgHolder = new MsgHolder(transactionId, getGlobalResponseTimeout());
        messageHolderStore.put(transactionId, msgHolder);
        msgHolder.setTimeoutFuture(executorService, () -> responseTimeout(transactionId));
        CompletableFuture<T> future = new CompletableFuture<>();
        msgHolder.setListener(new FutureMsgListener(future));
        return future;
    }

    @Override
    public void responseArrive(int transactionId, ModbusRspFrame<?> responseFrame) {
        MsgHolder msgHold = messageHolderStore.remove(transactionId);
        if(msgHold != null) {
            msgHold.cancelTimeoutFuture();
            executorService.execute(() -> msgHold.getListener().onSuccess(responseFrame.getBody()));
        } else {
            log.warn("modbus request is missing. transactionIdentifier:{} msg:{}", transactionId, responseFrame.getBody());
        }
    }

    @Override
    public void responseTimeout(int transactionId) {
        MsgHolder msgHolder = messageHolderStore.remove(transactionId);
        if(msgHolder != null) {
            String errorMsg = String.format("Modbus response is timeout. transactionIdentifier:{%s}", transactionId);
            executorService.execute(() -> msgHolder.getListener().onError(new TimeoutException(errorMsg)));
        }
    }

    protected int getGlobalResponseTimeout() {
        return globalResponseTimeout;
    }
}
